package com.enjoy.rocketmq.config;


import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.apache.rocketmq.spring.support.RocketMQUtil;
import org.springframework.messaging.Message;
import org.springframework.messaging.converter.StringMessageConverter;

import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;

public class MyTransactionImpl implements RocketMQLocalTransactionListener {

    private Map<Object, Message<?>> localTrans = new ConcurrentHashMap<>();

    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object args) {
        Object transId = message.getHeaders().get(RocketMQHeaders.PREFIX + RocketMQHeaders.TRANSACTION_ID);
        String destination = args.toString();
        localTrans.put(transId, message);
        System.out.println("执行本地事务 msg = " + message);
        // 转成RocketMQ的Message对象
        org.apache.rocketmq.common.message.Message msg = RocketMQUtil.convertToRocketMessage(new StringMessageConverter(), "UTF-8", destination, message);

        String tags = msg.getTags();

        if (tags.contains("TagA")) {
            return RocketMQLocalTransactionState.COMMIT;
        } else if (tags.contains("TagB")) {
            return RocketMQLocalTransactionState.ROLLBACK;
        } else {
            return RocketMQLocalTransactionState.UNKNOWN;
        }
    }

    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
        Object transId = message.getHeaders().get(RocketMQHeaders.PREFIX + RocketMQHeaders.TRANSACTION_ID);
        Message<?> msg = localTrans.get(transId);
        // 这里能够获取到自定义的transaction_id属性
        System.out.println("checkLocalTransaction msg = " + msg);

        String tags = Objects.requireNonNull(msg.getHeaders().get(RocketMQHeaders.PREFIX + RocketMQHeaders.TAGS)).toString();
        if (tags.contains("TagC")) {
            return RocketMQLocalTransactionState.COMMIT;
        } else if (tags.contains("TagD")) {
            return RocketMQLocalTransactionState.ROLLBACK;
        } else {
            return RocketMQLocalTransactionState.UNKNOWN;
        }
    }
}
